package com.dky.satell.controller;

import com.dky.satell.dto.ResultDto;
import com.dky.satell.entity.MessageConfig;
//import com.dky.satell.kafka.acquisition.KafkaConsumerInit;
import com.dky.satell.kafka.producer.KafkaProducer;
import com.dky.satell.rocketmq.consumer.RocketMQConsumerInit;
import com.dky.satell.service.impl.MessageConfigServiceImpl;
import com.github.pagehelper.Page;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author mqc
 * @since 2021-04-25
 */
@Slf4j
@RestController
@RequestMapping("/gather/messageConfig")
public class MessageConfigController {

    @Resource
    private RocketMQConsumerInit rocketmqConsumerInit;
//
//    @Resource
//    private KafkaConsumerInit kafkaConsumerInit;

    @Resource
    private KafkaProducer kafkaProducer;

    @Resource
    private MessageConfigServiceImpl service;

    //消息队列查询
    @PostMapping("/list")
    public ResultDto findList() {
        List<MessageConfig> list = service.list();
        return ResultDto.success("成功！", list);
    }

    /**
     * 分页加查询加搜索
     *
     * @param maps
     * @return
     */
    @PostMapping("/list2")
    public ResultDto findMessageConfig(@RequestBody Map<String, Object> maps) {
        Integer pageNo = (Integer) maps.get("pageNo");
        Integer pageSize = (Integer) maps.get("pageSize");
        String name = (String) maps.get("name");
        Page<MessageConfig> messageConfigList = service.findMessageConfigList(pageNo, pageSize, name);
        HashMap<String, Object> datumMap = new HashMap<>();
        datumMap.put("total", messageConfigList.getTotal());
        datumMap.put("pageNo", messageConfigList.getPageNum());
        datumMap.put("pages", messageConfigList.getPages());
        datumMap.put("pageSize", messageConfigList.getPageSize());
        datumMap.put("rows", messageConfigList.getResult());
        List listIssueOrganization = service.findListIssueOrganization();
        datumMap.put("organization",listIssueOrganization);
        return ResultDto.success("", datumMap);
    }

    /**
     * 添加
     *
     * @param messageConfig
     * @return
     */
    @PostMapping("saveOrUpdate")
    public ResultDto saveMessageConfig(@RequestBody MessageConfig messageConfig) {
        Integer id = messageConfig.getId();
        boolean b = service.saveOrUpdate(messageConfig);
        if (b) {
//            if(id == null) {//连接信息编辑时禁用
//                switch (messageConfig.getType()) {
//                    case 1 :
//                        rocketmqConsumerInit.subscribe(messageConfig);
//                        break;
//                    case 2 :
//                        kafkaConsumerInit.subscribe(messageConfig);
//                        break;
//                    default:
//                        log.error("不支持的消息队列类型");
//                }
//            }
//            kafkaProducer.sendMessageConfig(messageConfig);
            return ResultDto.success("添加成功！",messageConfig);
        }
        if (messageConfig.getId() != null) {
            return ResultDto.fail("修改成功！",messageConfig);
        } else {
            return ResultDto.fail("失败",messageConfig);
        }

    }

    /**
     * 删除接口目前不是逻辑删除
     */
    @PostMapping("delete")
    public ResultDto findDelete(@RequestBody MessageConfig messageConfig) {
        Integer id = messageConfig.getId();
        boolean b = service.removeById(id);
        if (b) {
//            switch (messageConfig.getType()) {
//                case 1 :
//                    rocketmqConsumerInit.unsubscribe(messageConfig);
//                    break;
//                case 2 :
//                    kafkaConsumerInit.unsubscribe(messageConfig);
//                    break;
//                default:
//                    log.error("不支持的消息队列类型");
//            }
//            messageConfig.setDeleted((byte) 1);
//            kafkaProducer.sendMessageConfig(messageConfig);
            return ResultDto.success("成功！",messageConfig);
        } else {
            return ResultDto.fail("失败！",messageConfig);
        }
    }
}

